package util;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by zhang on 2017/3/11.
 *
 * zkServer只允许单链接访问
 *
 */
@SuppressWarnings("all")
public class ZkConnectPool {

    //变量
    private final int connectNum = 5;
    private final String ZK_ADDR ="localhost:2181";
    private final int OUT_TIME = 5000;
    private final int CHECK_CONNECT_VALIED_TIME = 30 ;
    private final int CHECK_FREE_CONNECT_TIME = 1 ;
    private final int GET_CONNECT_OUT_TIME = 5 ;
    public static final int NO_VERSION = -1 ;

    //常量
    private  final List<ZkConnect> list = Collections.synchronizedList(new ArrayList<ZkConnect>(connectNum));

    public synchronized void initConnectPool()throws Exception{
        final CountDownLatch countDownLatch =new CountDownLatch(connectNum);
        for (int i = 0 ; i < connectNum && Integer.valueOf(String.valueOf(countDownLatch.getCount()))  >-1 ; i++ )
            try {
                list.add(new ZkConnect( ZK_ADDR, OUT_TIME , (watchedEvent)-> {
                    if(watchedEvent.getState() == Watcher.Event.KeeperState.SyncConnected)
                        countDownLatch.countDown();
                    Watcher.Event.EventType event = watchedEvent.getType();
                    System.out.println(event+"  : "+watchedEvent.getPath());
                    //将发送到消息队列中
                    if (Watcher.Event.EventType.NodeDeleted == event){
                        System.out.println(" delete : "+watchedEvent.getPath());
                    }

                },Boolean.FALSE));
            }catch (Exception e){
                e.printStackTrace();
            }
        countDownLatch.await();
        System.out.println("-----------------线程池加载完毕--------------------------");
        checkConnectValiedTask();
    }

    public  void checkConnectValiedTask()throws Exception{
        AtomicInteger current = new AtomicInteger(0);
        int pollingInterval=CHECK_CONNECT_VALIED_TIME/connectNum*1000;
        System.out.println("-----------------执行轮训检测--------------------------");
        new Thread( () -> {
            CountDownLatch countDownLatch =new CountDownLatch(1);
            while (Boolean.TRUE){
                try {
                    Thread.sleep(pollingInterval);
                    if(current.get() == connectNum)
                        current.set(0);
                    else if(current.get() > connectNum)
                        current.set(current.get() % connectNum);
                    ZooKeeper zk=list.get(current.get());
                    if( zk != null && zk.getState().isAlive() ){
                        current.incrementAndGet();
                    } else{
                        list.set(current.get(), new ZkConnect( ZK_ADDR, OUT_TIME , (watchedEvent)-> {
                            if(watchedEvent.getState() == Watcher.Event.KeeperState.SyncConnected)
                                countDownLatch.countDown();
                        },Boolean.FALSE));
                        countDownLatch.await();
                        current.incrementAndGet();
                    }
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }).start();
    }


    public synchronized ZkConnect getZkConnect()throws Exception{
        Long startTime =System.currentTimeMillis();
        Optional<ZkConnect> optional=list.stream().filter(zkConnect -> !zkConnect.isIs_used()).findFirst();
        if(optional !=null  && optional.isPresent() && optional.get() !=null){
            optional.get().setIs_used(Boolean.TRUE);
            return optional.get();
        }
        while (Boolean.TRUE){
            if (System.currentTimeMillis() - GET_CONNECT_OUT_TIME*1000 > startTime)
                throw new  Exception("GET_CONNECT_OUT_TIME");
            Thread.sleep(CHECK_FREE_CONNECT_TIME*1000);
            optional=list.stream().filter(zkConnect -> !zkConnect.isIs_used()).findFirst();
            if(optional !=null  && optional.isPresent() && optional.get() !=null){
                optional.get().setIs_used(Boolean.TRUE);
                return optional.get();
            }
        }
        return null;
    }


    public synchronized void releaseZkConnect(ZkConnect zk)throws Exception{
        zk.setIs_used(Boolean.FALSE);
    }







    public static void main(String[] arys)throws Exception{
        ZkConnectPool pool =new ZkConnectPool();
        pool.initConnectPool();
        pool.releaseZkConnect(pool.getZkConnect());
        pool.releaseZkConnect(pool.getZkConnect());
        pool.releaseZkConnect(pool.getZkConnect());
        pool.releaseZkConnect(pool.getZkConnect());
        pool.releaseZkConnect(pool.getZkConnect());

        ZkConnect connect=null;
        //永久节点
        //pool.getZkConnect().create("/zhang","juntao".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);

        //临时节点 本次会话内对所有client有效（分布式锁）
        pool.getZkConnect().create("/zhang/tmp","juntao".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL);

        System.out.println(new String(pool.getZkConnect().getData("/zhang/tmp",false,null)));

        //乐观锁   zkDataVersion默认为0，每次setData、delete传入的version会和zkServer中的version比较，相同更新，不相同抛出异常,版本号修改的次数
        connect=pool.getZkConnect();
        connect.setData("/zhang/tmp","AFDJKJKDJKFS".getBytes(),0);
        pool.releaseZkConnect(connect);

        connect= pool.getZkConnect();
        connect.delete("/zhang/tmp",1);
        pool.releaseZkConnect(connect);

        new Thread( () ->{
            try {
                ZkConnect connect1= pool.getZkConnect();
                connect1.create("/zhang/tmp","juntao".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL);
                pool.releaseZkConnect(connect1);
            }catch (Exception e){
                e.printStackTrace();
            }
        }).start();
        System.out.println(pool.getZkConnect().exists("/zhang/tmp2",false));
    }


























}
